#include "session.h"
#include "src/define.h"
#include "worker.h"
#include "core.h"
#include "log.h"
#include "socks5proto.h"
#include <string.h>
#include <malloc.h>

static void _OnSessionCheck(uv_timer_t *handle)
{
    LOG_DEBUG("_OnSessionCheck");
    Session *pSession = (Session *)uv_handle_get_data((uv_handle_t *)handle);
    if (pSession == NULL)
    {
        uv_timer_stop(handle);
        return;
    }

    pSession->OnSessionCheck();
}

static void _OnAllocBuff(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
    Session *pSession = (Session *)uv_handle_get_data(handle);

    buf->base = pSession->m_cmdRecvBuf + pSession->m_cmdRecvLen;
    buf->len = CMD_BUFF_SIZE - pSession->m_cmdRecvLen;
    return;
}

static void _OnReadData(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf)
{
    LOG_DEBUG("_OnReadData");
    Session *pSession = (Session *)uv_handle_get_data((uv_handle_t *)stream);
    if (nread >= 0 && pSession != NULL)
    {
        pSession->HandCommand(nread);
        return;
    }
    pSession->Close(true);
    return;
}

static void _OnWriteData(uv_write_t *req, int status)
{
    LOG_DEBUG("_OnWriteData");
    Session *pSession = (Session *)uv_handle_get_data((uv_handle_t *)req);
    if (pSession == NULL)
    {
        free(req);
        return;
    }
    if (status != 0)
    {
        LOG_ERROR("Session: %08x Send Message Failed", pSession->m_sessionId);
        pSession->Close(true);
    }
    free(req);
    return;
}

static void _OnSessionClose(uv_handle_t *handle)
{
    LOG_DEBUG("_OnSessionClose");
    Session *pSession = (Session *)uv_handle_get_data(handle);
    if (pSession == NULL)
    {
        return;
    }
    if (pSession->m_connectBuf.base)
    {
        LOG_DEBUG("_OnSessionClose m_connectBuf");
        free(pSession->m_connectBuf.base);
        pSession->m_connectBuf.base = NULL;
    }
    if (pSession->m_targetBuf.base)
    {
        LOG_DEBUG("_OnSessionClose m_targetBuf");
        free(pSession->m_targetBuf.base);
        LOG_DEBUG("_OnSessionClose m_targetBuf");
        pSession->m_targetBuf.base = NULL;
    }
    LOG_DEBUG("_OnSessionClose pSession");
    delete pSession;
}

static void _OnResolvedTarget(uv_getaddrinfo_t *req, int status, struct addrinfo *res)
{
    Session *pSession = (Session *)uv_req_get_data((uv_req_t *)req);
    if (status == 0)
    {
        pSession->DoTcpConnectTarget(res->ai_addr);
        uv_freeaddrinfo(res);
    }
    else
    {
        LOG_ERROR("getaddrinfo error: %s", uv_strerror(status));
        pSession->Close(true);
    }
}

static void _OnUdpResolvedTarget(uv_getaddrinfo_t *req, int status, struct addrinfo *res)
{
    LOG_DEBUG("_OnUdpResolvedTarget");
    Session *pSession = (Session *)uv_req_get_data((uv_req_t *)req);
    if (status == 0)
    {
        pSession->DoUdpSendToTarget(res->ai_addr);
        uv_freeaddrinfo(res);
    }
    else
    {
        LOG_ERROR("getaddrinfo error: %s", uv_strerror(status));
        pSession->Close(true);
    }
}

static void _OnTcpConnectTarget(uv_connect_t *req, int status)
{
    Session *pSession = (Session *)uv_req_get_data((uv_req_t *)req);
    if (status == 0)
    {
        pSession->SendCMDResp(pSession->m_cmdSendBuf, SOCKS5_VERSION, SOCKS5_SUCCEEDED, 0, SOCKS5_ATYP_IPV4, g_PublicIP, 0);
        return;
    }
    else
    {
        LOG_ERROR("OnConnectTarget error: %s", uv_strerror(status));
        pSession->SendCMDResp(pSession->m_cmdSendBuf, SOCKS5_VERSION, SOCKS5_CONNECTION_REFUSED, 0, SOCKS5_ATYP_IPV4, 0, 0);
        pSession->Close(true);
    }
}

static void _OnSendCMDResp(uv_write_t *req, int status)
{
    Session *pSession = (Session *)uv_req_get_data((uv_req_t *)req);
    if (status == 0)
    {
        if (pSession->m_Model == SOCKS_CMD_CONNECT)
        {
            pSession->DoTcpConnectedTarget();
        }
        else if (pSession->m_Model == SOCKS5_CMD_UDP)
        {
            pSession->DoUdpConnectedTarget();
        }
        else
        {
            pSession->Close(true);
        }
        return;
    }
    else
    {
        LOG_ERROR("OnSendCMDResp error: %s", uv_strerror(status));
        pSession->Close(true);
    }
}

static void _OnTcpAllocBuffConnect(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
    Session *pSession = (Session *)uv_handle_get_data(handle);

    buf->base = pSession->m_connectBuf.base + pSession->m_connectRecvLen;
    buf->len = TCP_BUFF_SIZE - pSession->m_connectRecvLen;
    return;
}

static void _OnTcpAllocBuffTarget(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
    Session *pSession = (Session *)uv_handle_get_data(handle);

    buf->base = pSession->m_targetBuf.base + pSession->m_targetRecvLen;
    buf->len = TCP_BUFF_SIZE - pSession->m_targetRecvLen;
    return;
}

static void _OnTcpReadConnect(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf);
static void _OnTcpWriteDataTarget(uv_write_t *req, int status);
static void _OnTcpWriteDataConnect(uv_write_t *req, int status);
static void _OnTcpReadTarget(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf);

static void _OnTcpWriteDataTarget(uv_write_t *req, int status)
{
    LOG_DEBUG("_OnTcpWriteDataTarget");
    Session *pSession = (Session *)uv_req_get_data((uv_req_t *)req);
    if (pSession == NULL)
    {
        free(req);
        return;
    }
    if (status != 0)
    {
        LOG_ERROR("Session: %08x Send Message Failed", pSession->m_sessionId);
        pSession->Close(true);
    }
    if (pSession->m_connectRecvLen == TCP_BUFF_SIZE)
    {
        uv_read_start((uv_stream_t *)&pSession->m_connectTcp, _OnTcpAllocBuffConnect, _OnTcpReadConnect);
    }
    pSession->m_connectRecvLen = 0;
    free(req);
    return;
}

static void _OnTcpReadConnect(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf)
{
    LOG_DEBUG("_OnTcpReadConnect %d", (int)nread);
    Session *pSession = (Session *)uv_handle_get_data((uv_handle_t *)stream);
    if (nread >= 0 && pSession != NULL)
    {
        pSession->m_lastRecvTick = uv_now(pSession->m_worker->GetLoop());
        uv_write_t *write_req = (uv_write_t *)calloc(1, sizeof(uv_write_t));
        uv_req_set_data((uv_req_t *)write_req, pSession);
        if (write_req != NULL)
        {
            uv_write(write_req, (uv_stream_t *)pSession->m_targetTcp, &pSession->m_connectBuf, 1, _OnTcpWriteDataTarget);
            return;
        }
        if (pSession->m_connectRecvLen == TCP_BUFF_SIZE)
        {
            uv_read_stop((uv_stream_t *)&pSession->m_connectTcp);
        }
    }
    LOG_DEBUG("_OnTcpReadConnect error:%s", uv_strerror(nread));
    pSession->Close(true);
    return;
}

static void _OnTcpWriteDataConnect(uv_write_t *req, int status)
{
    LOG_DEBUG("_OnTcpWriteDataConnect");
    Session *pSession = (Session *)uv_req_get_data((uv_req_t *)req);
    if (pSession == NULL)
    {
        free(req);
        return;
    }
    if (status != 0)
    {
        LOG_ERROR("Session: %08x Send Message Failed", pSession->m_sessionId);
        pSession->Close(true);
    }
    if (pSession->m_targetRecvLen == TCP_BUFF_SIZE)
    {
        uv_read_start((uv_stream_t *)pSession->m_targetTcp, _OnTcpAllocBuffTarget, _OnTcpReadTarget);
    }
    pSession->m_targetRecvLen = 0;
    free(req);
    return;
}

static void _OnTcpReadTarget(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf)
{
    LOG_DEBUG("_OnTcpReadTarget %d", (int)nread);
    Session *pSession = (Session *)uv_handle_get_data((uv_handle_t *)stream);
    if (nread >= 0 && pSession != NULL)
    {
        pSession->m_lastRecvTick = uv_now(pSession->m_worker->GetLoop());
        uv_write_t *write_req = (uv_write_t *)calloc(1, sizeof(uv_write_t));
        uv_req_set_data((uv_req_t *)write_req, pSession);
        if (write_req != NULL)
        {
            uv_write(write_req, (uv_stream_t *)&pSession->m_connectTcp, &pSession->m_targetBuf, 1, _OnTcpWriteDataConnect);
            return;
        }
        if (pSession->m_targetRecvLen == TCP_BUFF_SIZE)
        {
            uv_read_stop((uv_stream_t *)pSession->m_targetTcp);
        }
    }
    LOG_DEBUG("_OnTcpReadTarget error:%s", uv_strerror(nread));
    pSession->Close(true);
    return;
}

static void _OnUdpAllocBuffConnect(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
    Session *pSession = (Session *)uv_handle_get_data(handle);
    LOG_DEBUG("_OnUdpAllocBuffConnect suggested_size %ld m_udpHeaderLen %u", suggested_size, pSession->m_udpHeaderLen);
    buf->base = pSession->m_connectBuf.base;
    buf->len = UDP_BUFF_SIZE;
    return;
}

static void _OnUdpAllocBuffTarget(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
    Session *pSession = (Session *)uv_handle_get_data(handle);
    LOG_DEBUG("_OnUdpAllocBuffTarget suggested_size %ld m_udpHeaderLen %u", suggested_size, pSession->m_udpHeaderLen);
    buf->base = pSession->m_targetBuf.base + pSession->m_udpHeaderLen;
    buf->len = UDP_BUFF_SIZE - pSession->m_udpHeaderLen;

    return;
}

static void _OnUdpWriteDataTarget(uv_udp_send_t *req, int status)
{
    LOG_DEBUG("_OnUdpWriteDataTarget");
    Session *pSession = (Session *)uv_req_get_data((uv_req_t *)req);
    if (pSession == NULL)
    {
        free(req);
        return;
    }
    if (status != 0)
    {
        LOG_ERROR("Session: %08x Send Message Failed", pSession->m_sessionId);
        pSession->Close(true);
    }
    pSession->m_connectRecvLen = 0;
    free(req);
    return;
}

static void _OnUdpWriteDataConnect(uv_udp_send_t *req, int status)
{
    LOG_DEBUG("_OnUdpWriteDataConnect");
    Session *pSession = (Session *)uv_req_get_data((uv_req_t *)req);
    if (pSession == NULL)
    {
        free(req);
        return;
    }
    if (status != 0)
    {
        LOG_ERROR("Session: %08x Send Message Failed", pSession->m_sessionId);
        pSession->Close(true);
    }
    pSession->m_targetRecvLen = 0;
    free(req);
    return;
}

static void _OnUdpReadConnect(uv_udp_t *handle, ssize_t nread, const uv_buf_t *buf, const sockaddr *addr, unsigned flags)
{
    LOG_DEBUG("_OnUdpReadConnect %d", (int)nread);
    Session *pSession = (Session *)uv_handle_get_data((uv_handle_t *)handle);
    if (nread < 0)
    {
        LOG_DEBUG("_OnUdpReadConnect error:%s", uv_strerror(nread));
        pSession->Close(true);
        return;
    }
    else if (nread == 0)
    {
        return;
    }
    pSession->m_lastRecvTick = uv_now(pSession->m_worker->GetLoop());
    pSession->m_connectRecvLen = nread;
    if (nread < 4)
    {
        LOG_DEBUG("package too short");
        return;
    }

    if (!pSession->m_hastarget)
    {
        char ip[INET6_ADDRSTRLEN];
        uv_ip4_name((const struct sockaddr_in *)addr, ip, sizeof(ip));
        LOG_DEBUG("_OnUdpReadConnect Resolved: %s %u\n", ip, ((struct sockaddr_in *)addr)->sin_port);
        memcpy(&pSession->m_connectAddr, addr, sizeof(struct sockaddr));
        uint8_t atype = buf->base[3];
        char *cmdPos = buf->base;
        if (atype == SOCKS5_ATYP_IPV4)
        {
            if (nread < 10)
            {
                return;
            }
            pSession->m_udpHeaderLen = 10;
            snprintf(pSession->m_targetInfo, TARGET_INFO_SIZE, "%u.%u.%u.%u", (uint8_t)cmdPos[4], (uint8_t)cmdPos[5],
                     (uint8_t)cmdPos[6], (uint8_t)cmdPos[7]);
            pSession->m_targetPort = *(uint16_t *)(cmdPos + 8);
        }
        else if (atype == SOCKS5_ATYP_DOMAINNAME)
        {
            if (nread < 5)
            {
                return;
            }
            uint8_t domainLen = cmdPos[4];
            if (nread < 7 + domainLen)
            {
                return;
            }
            memcpy(pSession->m_targetInfo, cmdPos + 5, domainLen);
            pSession->m_targetPort = *(uint16_t *)(cmdPos + 5 + domainLen);
            pSession->m_udpHeaderLen = 5 + domainLen + 2;
        }
        else
        {
            LOG_DEBUG("atype Not Supported");
            return;
        }
        uv_udp_recv_stop(&pSession->m_connectUdp);
        LOG_DEBUG("_OnUdpReadConnect targetInfo %s", pSession->m_targetInfo);
        memcpy(pSession->m_targetBuf.base, pSession->m_connectBuf.base, pSession->m_udpHeaderLen);
        struct addrinfo hints;
        hints.ai_family = PF_INET;
        hints.ai_socktype = SOCK_STREAM;
        hints.ai_protocol = IPPROTO_TCP;
        hints.ai_flags = 0;
        uv_req_set_data((uv_req_t *)&pSession->m_resolver, pSession);
        int r = uv_getaddrinfo(pSession->m_worker->GetLoop(), &pSession->m_resolver, _OnUdpResolvedTarget, pSession->m_targetInfo,
                               std::to_string(htons(pSession->m_targetPort)).c_str(), &hints);
        if (r)
        {
            LOG_ERROR("getaddrinfo [%s] call error %s", pSession->m_targetInfo, uv_err_name(r));
            return;
        }
    }
    else
    {
        LOG_DEBUG("send_buf targetUdp  %d", (int)pSession->m_connectRecvLen - pSession->m_udpHeaderLen);
        uv_buf_t send_buf = {pSession->m_connectBuf.base + pSession->m_udpHeaderLen, pSession->m_connectRecvLen - pSession->m_udpHeaderLen};
        auto send_req = (uv_udp_send_t *)calloc(1, sizeof(uv_udp_send_t));
        uv_req_set_data((uv_req_t *)send_req, pSession);
        if (send_req != NULL)
        {
            LOG_DEBUG("DoUdpSendToTarget len %ld", send_buf.len);
            uv_udp_send(send_req, &pSession->m_targetUdp, &send_buf, 1, (const sockaddr *)&pSession->m_targetAddr, _OnUdpWriteDataTarget);
            return;
        }
    }
}

static void _OnUdpReadTarget(uv_udp_t *handle, ssize_t nread, const uv_buf_t *buf, const sockaddr *addr, unsigned flags)
{
    LOG_DEBUG("_OnUdpReadTarget %d", (int)nread);
    auto pSession = (Session *)uv_handle_get_data((uv_handle_t *)handle);

    if (nread > 0)
    {
        pSession->m_lastRecvTick = uv_now(pSession->m_worker->GetLoop());
        if (!pSession->m_hastarget)
        {
            LOG_DEBUG("udp drop packet");
            return;
        }
        pSession->m_targetRecvLen = nread;
        uv_buf_t send_buf = {pSession->m_targetBuf.base, pSession->m_targetRecvLen + pSession->m_udpHeaderLen};
        auto send_req = (uv_udp_send_t *)calloc(1, sizeof(uv_udp_send_t));
        uv_req_set_data((uv_req_t *)send_req, pSession);
        if (send_req != nullptr)
        {
            LOG_DEBUG("DoUdpSendToConnect len %ld", send_buf.len);
            uv_udp_send(send_req, &pSession->m_connectUdp, &send_buf, 1, (const sockaddr *)&pSession->m_connectAddr, _OnUdpWriteDataConnect);
            return;
        }
    }
    else if (nread == 0)
    {
        return;
    }
    LOG_DEBUG("_OnUdpReadTarget error:%s", uv_strerror(nread));
    pSession->Close(true);
    return;
}

static void _OnUdpViaTcpAllocBuffConnect(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
    Session *pSession = (Session *)uv_handle_get_data(handle);
    LOG_DEBUG("_OnUdpViaTcpAllocBuffConnect suggested_size %ld m_udpHeaderLen %u", suggested_size, pSession->m_udpHeaderLen);
    buf->base = pSession->m_connectBuf.base + pSession->m_connectRecvLen;
    buf->len = UDP_BUFF_SIZE - pSession->m_connectRecvLen;
    return;
}

static void _OnUdpViaTcpAllocBuffTarget(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
    Session *pSession = (Session *)uv_handle_get_data(handle);

    buf->base = pSession->m_targetBuf.base + pSession->m_targetRecvLen;
    buf->len = UDP_BUFF_SIZE - pSession->m_targetRecvLen;
    return;
}

static void _OnUdpViaTcpWriteDataTarget(uv_write_t *req, int status)
{
    LOG_DEBUG("_OnTcpWriteDataTarget");
    Session *pSession = (Session *)uv_req_get_data((uv_req_t *)req);
    if (pSession == NULL)
    {
        free(req);
        return;
    }
    if (status != 0)
    {
        LOG_ERROR("Session: %08x Send Message Failed", pSession->m_sessionId);
        pSession->Close(true);
    }
    pSession->m_connectRecvLen = 0;
    free(req);
    return;
}

static void _OnUdpViaTcpReadConnect(uv_udp_t *handle, ssize_t nread, const uv_buf_t *buf, const sockaddr *addr, unsigned flags)
{
    LOG_DEBUG("_OnUdpViaTcpReadConnect %d", (int)nread);
    Session *pSession = (Session *)uv_handle_get_data((uv_handle_t *)handle);
    if (nread < 0)
    {
        LOG_DEBUG("_OnUdpViaTcpReadConnect error:%s", uv_strerror(nread));
        pSession->Close(true);
        return;
    }
    else if (nread == 0)
    {
        return;
    }
    pSession->m_lastRecvTick = uv_now(pSession->m_worker->GetLoop());
    pSession->m_connectRecvLen = nread;
    if (nread < 4)
    {
        LOG_DEBUG("package too short");
        return;
    }

    uint8_t atype = buf->base[3];
    char *cmdPos = buf->base;
    if (atype == SOCKS5_ATYP_IPV4)
    {
        if (nread < 10)
        {
            return;
        }
        pSession->m_udpHeaderLen = 10;
        if (!pSession->m_hastarget)
        {
            char ip[INET6_ADDRSTRLEN];
            uv_ip4_name((const struct sockaddr_in *)addr, ip, sizeof(ip));
            LOG_DEBUG("_OnUdpViaTcpReadConnect Resolved: %s %u\n", ip, ((struct sockaddr_in *)addr)->sin_port);
            memcpy(&pSession->m_connectAddr, addr, sizeof(struct sockaddr));
            snprintf(pSession->m_targetInfo, TARGET_INFO_SIZE, "%u.%u.%u.%u", (uint8_t)cmdPos[4], (uint8_t)cmdPos[5],
                        (uint8_t)cmdPos[6], (uint8_t)cmdPos[7]);
            pSession->m_targetPort = *(uint16_t *)(cmdPos + 8);
            pSession->m_hastarget = true;
        }
    }
    else if (atype == SOCKS5_ATYP_DOMAINNAME)
    {
        if (nread < 5)
        {
            return;
        }
        uint8_t domainLen = cmdPos[4];
        if (nread < 7 + domainLen)
        {
            return;
        }
        if (!pSession->m_hastarget)
        {
            char ip[INET6_ADDRSTRLEN];
            uv_ip4_name((const struct sockaddr_in *)addr, ip, sizeof(ip));
            LOG_DEBUG("_OnUdpViaTcpReadConnect : %s %u\n", ip, ((struct sockaddr_in *)addr)->sin_port);
            memcpy(&pSession->m_connectAddr, addr, sizeof(struct sockaddr));
            memcpy(pSession->m_targetInfo, cmdPos + 5, domainLen);
            pSession->m_targetPort = *(uint16_t *)(cmdPos + 5 + domainLen);
            pSession->m_udpHeaderLen = 5 + domainLen + 2;
            pSession->m_hastarget = true;
        }
    }
    else
    {
        LOG_DEBUG("atype Not Supported");
        return;
    }
    uint16_t *data_len = (uint16_t *)pSession->m_connectBuf.base;
    *data_len = nread;
    uv_buf_t send_buf = {pSession->m_connectBuf.base, (size_t)nread};
    uv_write_t *write_req = (uv_write_t *)calloc(1, sizeof(uv_write_t));
    uv_req_set_data((uv_req_t *)write_req, pSession);
    if (write_req != NULL)
    {
        LOG_DEBUG("DoUdpSendToConnect len %ld type %d", send_buf.len, pSession->m_targetTcp->type);
        uv_write(write_req, (uv_stream_t *)pSession->m_targetTcp, &send_buf, 1, _OnUdpViaTcpWriteDataTarget);
        return;
    } 
}

static void _OnUdpViaTcpReadTarget(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf)
{
    LOG_DEBUG("_OnUdpViaTcpReadTarget %d", (int)nread);
    Session *pSession = (Session *)uv_handle_get_data((uv_handle_t *)stream);
    if (nread >= 0 && pSession != NULL)
    {
        pSession->m_lastRecvTick = uv_now(pSession->m_worker->GetLoop());
        pSession->m_targetRecvLen += nread;
        uint16_t *data_len = (uint16_t *)pSession->m_targetBuf.base;
        LOG_DEBUG("_OnUdpViaTcpReadTarget data_len %u", *data_len);
        if (*data_len > pSession->m_targetRecvLen) {
            return;
        }
        *data_len = 0;
        uv_buf_t send_buf = {pSession->m_targetBuf.base, pSession->m_targetRecvLen};
        auto send_req = (uv_udp_send_t *)calloc(1, sizeof(uv_udp_send_t));
        uv_req_set_data((uv_req_t *)send_req, pSession);
        if (send_req != nullptr)
        {
            LOG_DEBUG("DoUdpSendToConnect len %ld", send_buf.len);
            uv_udp_send(send_req, &pSession->m_connectUdp, &send_buf, 1, (const sockaddr *)&pSession->m_connectAddr, nullptr);
            pSession->m_targetRecvLen = 0;
            return;
        }
    }
    LOG_DEBUG("_OnTcpReadTarget error:%s", uv_strerror(nread));
    pSession->Close(true);
    return;
}

bool Session::Init(Worker *worker, uv_stream_t *pServer, uint32_t sessionId, int commandTimeout, int idleTimeout)
{
    m_status = SESSION_STATUS_INIT;
    m_ipaddr = 0;
    m_isAuthNone = false;
    m_userName = "";
    m_password = "";
    m_lastRecvTick = 0;
    m_hastarget = false;
    m_cmdRecvLen = 0;
    m_targetType = 0;
    m_connectRecvLen = 0;
    m_targetRecvLen = 0;
    m_udpHeaderLen = 0;
    m_Model = 0;
    m_worker = worker;
    m_sessionId = sessionId;
    m_commandTimeout = commandTimeout;
    m_idleTimeout = idleTimeout;
    if (!m_worker->AddSessionCache(this))
    {
        LOG_ERROR("Session: %08x Add Cache Failed", m_sessionId);
        return false;
    }
    uv_tcp_init(m_worker->GetLoop(), &m_connectTcp);
    uv_udp_init(m_worker->GetLoop(), &m_connectUdp);
    uv_udp_init(m_worker->GetLoop(), &m_targetUdp);
    if (uv_accept(pServer, (uv_stream_t *)&m_connectTcp) == 0)
    {
        struct sockaddr addr;
        int addrlen = sizeof(struct sockaddr);
        // 获取客户端IP地址
        if (uv_tcp_getpeername(&m_connectTcp, (struct sockaddr *)&addr, &addrlen) == 0)
        {
            if (addr.sa_family == AF_INET)
            {
                struct sockaddr_in const *ipv4 = (struct sockaddr_in *)&addr;
                m_ipaddr = ipv4->sin_addr.s_addr;
            }
            else
            { 
                LOG_INFO("Session: %08x not support ipv6", m_sessionId);
                Close(false);
                return false;
            }
        }

        uv_timer_init(m_worker->GetLoop(), &m_sessionTimer);
        uv_handle_set_data((uv_handle_t *)&m_sessionTimer, this);
        uv_handle_set_data((uv_handle_t *)&m_connectTcp, this);
        m_lastRecvTick = uv_now(m_worker->GetLoop());
        uv_timer_start(&m_sessionTimer, _OnSessionCheck, 0, g_timerRepeat);
        uv_read_start((uv_stream_t *)&m_connectTcp, _OnAllocBuff, _OnReadData);
        m_status = SESSION_STATUS_CONN;
        LOG_DEBUG("Session::Init");
        return true;
    }
    Close(false);
    return false;
}

bool Socks5IsMethodExists(uint8_t nmethods, uint8_t method, char *buf)
{
    for (int i = 0; i < nmethods; i++)
    {
        if (method == buf[i])
        {
            return true;
        }
    }
    return false;
}

void Session::HandCommand(ssize_t nread)
{
    LOG_DEBUG("Session: %08x ::HandCommand: %d", m_sessionId, nread);
    m_lastRecvTick = uv_now(m_worker->GetLoop());
    if (m_status == SESSION_STATUS_FIN)
    {
        return;
    }
    m_cmdRecvLen += nread;
    LOG_DEBUG("Session: %08x ::m_cmdRecvLen %u", m_sessionId, m_cmdRecvLen);
    char *cmdPos = m_cmdRecvBuf;
    char *sendPos = m_cmdSendBuf;
    if (m_status == SESSION_STATUS_CONN)
    {
        LOG_DEBUG("Session: %08x ::m_cmdRecvLen %u", m_sessionId, m_cmdRecvLen);
        //  +----+----------+----------+
        //  |VER | NMETHODS | METHODS  |
        //  +----+----------+----------+
        //  | 1  |    1     | 1 to 255 |
        //  +----+----------+----------+
        //                  [          ]
        uint8_t version = cmdPos[0];
        if (version != SOCKS5_VERSION)
        {
            m_status = SESSION_STATUS_FIN;
            LOG_DEBUG("Session: %08x ::version: %u", m_sessionId, version);
            Close(true);
            return;
        }
        if (m_cmdRecvLen < 2)
        {
            goto fin;
        }
        uint8_t nmethods = cmdPos[1];
        if (m_cmdRecvLen < 2 + nmethods)
        {
            goto fin;
        }
        sendPos[0] = version;
        // 判断是否支持并回复
        if (Socks5IsMethodExists(nmethods, SOCKS5_AUTH, cmdPos + 2))
        {
            sendPos[1] = (uint8_t)SOCKS5_AUTH;
            m_status = SESSION_STATUS_AUTH;
            LOG_DEBUG("Session: %08x Auth", m_sessionId);
        }
        else if (Socks5IsMethodExists(nmethods, SOCKS5_AUTH_NONE, cmdPos + 2))
        {
            if (m_worker->DoAuth(m_userName, m_password, m_ipaddr))
            {
                sendPos[1] = (uint8_t)SOCKS5_AUTH_NONE;
                m_status = SESSION_STATUS_CMD;
                LOG_DEBUG("Session: %08x Auth Pass", m_sessionId);
            }
            else
            {
                LOG_INFO("Session: %08x Auth failed", m_sessionId);
                sendPos[1] = (uint8_t)SOCKS5_AUTH_UNACCEPTABLE;
                m_status = SESSION_STATUS_FIN;
            }
        }
        else
        {
            LOG_DEBUG("Session: %08x ::nmethod", m_sessionId);
            sendPos[1] = (uint8_t)SOCKS5_AUTH_UNACCEPTABLE;
            m_status = SESSION_STATUS_FIN;
        }
        // send resp
        SendResp(sendPos, 2);
        sendPos += 2;
        if (m_status == SESSION_STATUS_FIN)
        {
            Close(true);
            return;
        }
        cmdPos += (2 + nmethods);
        m_cmdRecvLen -= (2 + nmethods);
        LOG_DEBUG("Session: %08x ::m_cmdRecvLen %u", m_sessionId, m_cmdRecvLen);
    }
    if (m_status == SESSION_STATUS_AUTH)
    {
        LOG_DEBUG("Session: %08x ::m_cmdRecvLen %u", m_sessionId, m_cmdRecvLen);
        if (m_cmdRecvLen < 3)
        {
            goto fin;
        }
        //  +----+------+----------+------+----------+
        //  |VER | ULEN |  UNAME   | PLEN |  PASSWD  |
        //  +----+------+----------+------+----------+
        //  | 1  |  1   | 1 to 255 |  1   | 1 to 255 |
        //  +----+------+----------+------+----------+
        uint8_t authVersion = cmdPos[0];
        if (SOCKS5_AUTH_VERSION != authVersion)
        {
            m_status = SESSION_STATUS_FIN;
            Close(true);
            return;
        }
        uint8_t authUserNameLen = cmdPos[1];
        if (m_cmdRecvLen < 3 + authUserNameLen)
        {
            goto fin;
        }
        uint8_t authPasswordLen = cmdPos[2 + authUserNameLen];
        if (m_cmdRecvLen < 3 + authUserNameLen + authPasswordLen)
        {
            goto fin;
        }
        if (authUserNameLen > 0)
        {
            m_userName = std::string(cmdPos + 2, authUserNameLen);
        }
        if (authPasswordLen)
        {
            m_password = std::string(cmdPos + 3 + authUserNameLen, authPasswordLen);
        }
        //  +----+--------+
        //  |VER | STATUS |
        //  +----+--------+
        //  | 1  |   1    |
        //  +----+--------+
        sendPos[0] = authVersion;
        if (m_worker->DoAuth(m_userName, m_password, m_ipaddr))
        {
            sendPos[1] = SOCKS5_SUCCEEDED;
            m_status = SESSION_STATUS_CMD;
        }
        else
        {
            LOG_INFO("Session: %08x Auth failed", m_sessionId);
            sendPos[1] = SOCKS5_GENERAL_SOCKS_SERVER_FAILURE;
            m_status = SESSION_STATUS_FIN;
        }
        SendResp(sendPos, 2);
        sendPos += 2;
        if (m_status == SESSION_STATUS_FIN)
        {
            Close(true);
            return;
        }
        cmdPos += (3 + authUserNameLen + authPasswordLen);
        m_cmdRecvLen -= (3 + authUserNameLen + authPasswordLen);
    }

    if (m_status == SESSION_STATUS_CMD)
    {
        //  +----+-----+-------+------+----------+----------+
        //  |VER | CMD |  RSV  | ATYP | DST.ADDR | DST.PORT |
        //  +----+-----+-------+------+----------+----------+
        //  | 1  |  1  | X'00' |  1   | Variable |    2     |
        //  +----+-----+-------+------+----------+----------+
        LOG_DEBUG("Session: %08x ::m_cmdRecvLen %u", m_sessionId, m_cmdRecvLen);
        if (m_cmdRecvLen < 6)
        {
            goto fin;
        }
        LOG_DEBUG("Session: %08x ::cmd %u %u %u %u", m_sessionId, cmdPos[0], cmdPos[1], cmdPos[2], cmdPos[3]);
        uint8_t connVersion = cmdPos[0];
        if (connVersion != SOCKS5_VERSION)
        {
            m_status = SESSION_STATUS_FIN;
            LOG_DEBUG("Session: %08x ::Cmd %u", m_sessionId, connVersion);
            Close(true);
            return;
        }
        uint8_t atype = cmdPos[3];
        if (atype == SOCKS5_ATYP_IPV4)
        {
            if (m_cmdRecvLen < 6 + 4)
            {
                goto fin;
            }
        }
        else if (atype == SOCKS5_ATYP_DOMAINNAME)
        {
            uint8_t domainLen = cmdPos[4];
            if (m_cmdRecvLen < 6 + 1 + domainLen)
            {
                goto fin;
            }
        }
        else
        {
            LOG_INFO("Session: %08x not support atype: %u", m_sessionId, atype);
            m_status = SESSION_STATUS_FIN;
            Close(true);
            return;
        }
        uv_read_stop((uv_stream_t *)&m_connectTcp);
        uint8_t cmd = cmdPos[1];
        LOG_DEBUG("Session: %08x ::cmd %u", m_sessionId, cmd);
        if (cmd == SOCKS_CMD_CONNECT)
        {
            DoTcpCommand(cmdPos);
        }
        else if (cmd == SOCKS5_CMD_UDP)
        {
            DoUdpCommand(cmdPos);
        }
        else
        {
            LOG_INFO("Session: %08x not support cmd: %u", m_sessionId, cmd);
            m_status = SESSION_STATUS_FIN;
            Close(true);
            return;
        }
    }
fin:
    if (m_cmdRecvLen > 0 && m_cmdRecvBuf != cmdPos)
    {
        memmove(m_cmdRecvBuf, cmdPos, m_cmdRecvLen);
    }
}

void Session::OnSessionCheck()
{
    if (m_status != SESSION_STATUS_RELAY)
    {
        if (m_lastRecvTick + m_commandTimeout < uv_now(m_worker->GetLoop()))
        {
            LOG_DEBUG("Session: %08x command Timeout", m_sessionId);
            Close(true);
            return;
        }
    }
    else
    {
        if (m_lastRecvTick + m_idleTimeout < uv_now(m_worker->GetLoop()))
        {
            LOG_DEBUG("Session: %08x Idle Timeout", m_sessionId);
            Close(true);
            return;
        }
    }
}

void Session::Close(bool cache)
{
    LOG_DEBUG("Session: %08x Close", m_sessionId);
    if (cache)
    {
        m_worker->DelSessionCache(this);
    }
    if (uv_is_active((uv_handle_t *)&m_sessionTimer))
    {
        LOG_DEBUG("Session Close uv_timer_stop");
        uv_timer_stop(&m_sessionTimer);
    }
    if (!uv_is_closing((uv_handle_t *)&m_targetUdp))
    {
        LOG_DEBUG("Session Close m_targetUdp");
        uv_close((uv_handle_t *)&m_targetUdp, NULL);
    }
    if (!uv_is_closing((uv_handle_t *)&m_connectUdp))
    {
        LOG_DEBUG("Session Close m_connectUdp");
        uv_close((uv_handle_t *)&m_connectUdp, NULL);
    }
    if (m_targetTcp != nullptr && !uv_is_closing((uv_handle_t *)m_targetTcp))
    {
        LOG_DEBUG("Session Close m_targetTcp");
        uv_close((uv_handle_t *)m_targetTcp, NULL);
    }
    if (!uv_is_closing((uv_handle_t *)&m_connectTcp))
    {
        LOG_DEBUG("Session Close m_connectTcp");
        uv_close((uv_handle_t *)&m_connectTcp, _OnSessionClose);
    }
}

void Session::DoTcpCommand(char *cmdPos)
{
    m_hastarget = true;
    m_Model = SOCKS_CMD_CONNECT;
    m_targetType = cmdPos[3];
    if (m_targetType == SOCKS5_ATYP_IPV4)
    {
        snprintf(m_targetInfo, TARGET_INFO_SIZE, "%u.%u.%u.%u", (uint8_t)cmdPos[4], (uint8_t)cmdPos[5], (uint8_t)cmdPos[6], (uint8_t)cmdPos[7]);
        m_targetPort = *(uint16_t *)(cmdPos + 8);
    }
    else if (m_targetType == SOCKS5_ATYP_DOMAINNAME)
    {
        uint8_t domainLen = cmdPos[4];
        memcpy(m_targetInfo, cmdPos + 5, domainLen);
        m_targetPort = *(uint16_t *)(cmdPos + 5 + domainLen);
    }
    uint8_t *ipStr = (uint8_t *)&m_ipaddr;
    LOG_INFO("Client [%u.%u.%u.%u] Tcp Proxy targetInfo[%s:%u]", ipStr[0], ipStr[1], ipStr[2], ipStr[3], m_targetInfo, htons(m_targetPort));
    if (m_worker->GetType() == WORKER_TYPE_PROXY)
    {
        struct addrinfo hints;
        hints.ai_family = PF_INET;
        hints.ai_socktype = SOCK_STREAM;
        hints.ai_protocol = IPPROTO_TCP;
        hints.ai_flags = 0;
        uv_req_set_data((uv_req_t *)&m_resolver, this);
        int r = uv_getaddrinfo(m_worker->GetLoop(), &m_resolver, _OnResolvedTarget, m_targetInfo,
                               std::to_string(htons(m_targetPort)).c_str(), &hints);
        if (r)
        {
            LOG_ERROR("getaddrinfo [%s] call error %s", m_targetInfo, uv_err_name(r));
            return;
        }
    }
    else if (m_worker->GetType() == WORKER_TYPE_SPIDER)
    {
        char buff[CMD_BUFF_SIZE];
        MessageTcp *msg = (MessageTcp *)buff;
        uint16_t msgLen = sizeof(MessageTcp);
        msg->header.type = MESSAGE_TCP_REQ;
        msg->eagleAddr = g_PublicIP;
        msg->eaglePort = m_worker->GetSpiderPort();
        msg->sessionId = m_sessionId;
        msg->targetPort = m_targetPort;
        msg->targetLen = strlen(m_targetInfo);
        memcpy(msg->target, m_targetInfo, msg->targetLen);
        msgLen += msg->targetLen;
        Core::GetInstance().GetMaster()->DoMsgToSpiderNode(m_userName, buff, msgLen);
        LOG_DEBUG("DoMsgToSpiderNode %s => %s:%d", m_userName.c_str(), m_targetInfo, htons(m_targetPort));
    }
    else
    {
        Close(true);
    }
}

void Session::DoTcpConnectTarget(struct sockaddr *target)
{
    m_targetTcp = (uv_tcp_t *)calloc(1, sizeof(uv_tcp_t));
    if (m_targetTcp)
    {
        uv_tcp_init(m_worker->GetLoop(), m_targetTcp);
        uv_handle_set_data((uv_handle_t *)&m_targetUdp, this);
        uv_req_set_data((uv_req_t *)&m_targetConn, this);
        uv_tcp_connect(&m_targetConn, m_targetTcp, target, _OnTcpConnectTarget);
    }
}

void Session::DoTcpConnectedTarget()
{
    m_connectBuf.base = (char *)calloc(TCP_BUFF_SIZE, sizeof(char));
    if (!m_connectBuf.base)
    {
        Close(true);
    }
    m_connectBuf.len = TCP_BUFF_SIZE;
    m_targetBuf.base = (char *)calloc(TCP_BUFF_SIZE, sizeof(char));
    if (!m_targetBuf.base)
    {
        Close(true);
    }
    m_targetBuf.len = TCP_BUFF_SIZE;

    uv_handle_set_data((uv_handle_t *)&m_connectTcp, this);
    uv_handle_set_data((uv_handle_t *)m_targetTcp, this);
    uv_read_start((uv_stream_t *)&m_connectTcp, _OnTcpAllocBuffConnect, _OnTcpReadConnect);
    uv_read_start((uv_stream_t *)m_targetTcp, _OnTcpAllocBuffTarget, _OnTcpReadTarget);
    m_status = SESSION_STATUS_RELAY;
}

void Session::DoUdpCommand(char *cmdPos)
{
    m_Model = SOCKS5_CMD_UDP;
    struct sockaddr_in addr;
    uv_ip4_addr("0.0.0.0", 0, &addr);
    uv_udp_bind(&m_connectUdp, (const struct sockaddr *)&addr, 0);
    int namelen = sizeof(m_ss);
    struct sockaddr_in *sin = NULL;
    if (uv_udp_getsockname(&m_connectUdp, (struct sockaddr *)&m_ss, &namelen) == 0)
    {
        sin = (struct sockaddr_in *)&m_ss;
        LOG_DEBUG("Udp Listening on port %d", ntohs(sin->sin_port));
    }
    else
    {
        LOG_ERROR("Session: %08x get Udp Port failed", m_sessionId);
        Close(true);
        return;
    }
    if (m_worker->GetType() == WORKER_TYPE_PROXY)
    {
        SendCMDResp(m_cmdSendBuf, SOCKS5_VERSION, SOCKS5_SUCCEEDED, 0, SOCKS5_ATYP_IPV4, g_PublicIP, ((struct sockaddr_in *)&m_ss)->sin_port);
    }
    else if (m_worker->GetType() == WORKER_TYPE_SPIDER)
    {
        char buff[CMD_BUFF_SIZE];
        MessageUdp *msg = (MessageUdp *)buff;
        uint16_t msgLen = sizeof(MessageUdp);
        msg->header.type = MESSAGE_UDP_REQ;
        msg->eagleAddr = g_PublicIP;
        msg->eaglePort = m_worker->GetSpiderPort();
        msg->sessionId = m_sessionId;
        Core::GetInstance().GetMaster()->DoMsgToSpiderNode(m_userName, buff, msgLen);
        LOG_DEBUG("DoMsgToSpiderNode %s udp", m_userName.c_str());
    }
    else
    {
        Close(true);
    }
}

void Session::DoUdpConnectedTarget()
{
    LOG_DEBUG("DoUdpConnectedTarget");
    m_connectBuf.base = (char *)calloc(UDP_BUFF_SIZE, sizeof(char));
    if (m_connectBuf.base == NULL)
    {
        Close(true);
        return;
    }
    m_connectBuf.len = UDP_BUFF_SIZE;
    m_targetBuf.base = (char *)calloc(UDP_BUFF_SIZE, sizeof(char));
    if (m_targetBuf.base == NULL)
    {
        Close(true);
        return;
    }
    m_targetBuf.len = UDP_BUFF_SIZE;
    if (m_worker->GetType() == WORKER_TYPE_PROXY)
    {
        LOG_DEBUG("DoUdpConnectedTarget WORKER_TYPE_PROXY");
        uv_handle_set_data((uv_handle_t *)&m_connectUdp, this);
        uv_handle_set_data((uv_handle_t *)&m_targetUdp, this);
        uv_udp_recv_start(&m_connectUdp, _OnUdpAllocBuffConnect, _OnUdpReadConnect);
        uv_udp_recv_start(&m_targetUdp, _OnUdpAllocBuffTarget, _OnUdpReadTarget);
    }
    else if (m_worker->GetType() == WORKER_TYPE_SPIDER)
    {
        LOG_DEBUG("DoUdpConnectedTarget WORKER_TYPE_SPIDER");
        uv_handle_set_data((uv_handle_t *)&m_connectUdp, this);
        uv_handle_set_data((uv_handle_t *)m_targetTcp, this);
        uv_read_start((uv_stream_t *)m_targetTcp, _OnUdpViaTcpAllocBuffTarget, _OnUdpViaTcpReadTarget);
        uv_udp_recv_start(&m_connectUdp, _OnUdpViaTcpAllocBuffConnect, _OnUdpViaTcpReadConnect);        
    }
    else
    {
        Close(true);
    }
    m_status = SESSION_STATUS_RELAY;
}

void Session::DoUdpSendToTarget(struct sockaddr *target)
{
    LOG_DEBUG("DoUdpSendToTarget");
    m_hastarget = true;
    char ip[INET6_ADDRSTRLEN];
    uv_ip4_name((struct sockaddr_in *)target, ip, sizeof(ip));
    LOG_DEBUG("_OnUdpResolvedTarget Resolved: %s %u", ip, ntohs(((struct sockaddr_in *)target)->sin_port));
    memcpy(&m_targetAddr, target, sizeof(struct sockaddr));

    uv_buf_t send_buf = {m_connectBuf.base + m_udpHeaderLen, m_connectRecvLen - m_udpHeaderLen};
    uv_udp_send_t *send_req = (uv_udp_send_t *)calloc(1, sizeof(uv_udp_send_t));
    uv_req_set_data((uv_req_t *)send_req, this);
    if (send_req != NULL)
    {
        LOG_DEBUG("DoUdpSendToTarget len %ld", send_buf.len);
        uv_udp_send(send_req, &m_targetUdp, &send_buf, 1, &m_targetAddr, _OnUdpWriteDataTarget);
        uv_udp_recv_start(&m_connectUdp, _OnUdpAllocBuffConnect, _OnUdpReadConnect);
        return;
    }
    Close(true);
}

void Session::SendResp(char *resp, uint16_t respLen)
{
    uv_write_t *writeReq = (uv_write_t *)calloc(1, sizeof(uv_write_t));
    if (!writeReq)
    {
        LOG_ERROR("Session: %08x calloc write Req Failed", m_sessionId);
        Close(true);
        return;
    }
    uv_handle_set_data((uv_handle_t *)writeReq, (void *)this);
    uv_buf_t buf = uv_buf_init((char *)resp, respLen);
    if (uv_write(writeReq, (uv_stream_t *)&m_connectTcp, &buf, 1, _OnWriteData) < 0)
    {
        LOG_ERROR("Session: %08x write Req Failed", m_sessionId);
        free(writeReq);
        Close(true);
    }
}

void Session::SendCMDResp(char *resp, uint8_t ver, uint8_t rep, uint8_t rsv, uint8_t atype, uint32_t addr, uint16_t port)
{
    uint8_t pos = 0;
    resp[pos++] = ver;
    resp[pos++] = rep;
    resp[pos++] = rsv;
    resp[pos++] = atype;
    memcpy(resp + pos, &addr, sizeof(uint32_t));
    pos += sizeof(uint32_t);
    memcpy(resp + pos, &port, sizeof(uint16_t));
    pos += sizeof(uint16_t);

    uv_write_t *writeReq = (uv_write_t *)calloc(1, sizeof(uv_write_t));
    if (!writeReq)
    {
        LOG_ERROR("Session: %08x calloc write Req Failed", m_sessionId);
        Close(true);
        return;
    }
    uv_handle_set_data((uv_handle_t *)writeReq, (void *)this);
    uv_buf_t buf = uv_buf_init((char *)resp, pos);
    if (uv_write(writeReq, (uv_stream_t *)&m_connectTcp, &buf, 1, _OnSendCMDResp) < 0)
    {
        LOG_ERROR("Session: %08x write Req Failed", m_sessionId);
        free(writeReq);
        Close(true);
    }
}

void Session::AttachTarget(uv_tcp_t *target)
{
    LOG_DEBUG("Session AttachTarget");
    m_targetTcp = target;
    if (m_Model == SOCKS_CMD_CONNECT)
    {
        uv_handle_set_data((uv_handle_t *)m_targetTcp, this);
        SendCMDResp(m_cmdSendBuf, SOCKS5_VERSION, SOCKS5_SUCCEEDED, 0, SOCKS5_ATYP_IPV4, g_PublicIP, 0);
    }
    else if (m_Model == SOCKS5_CMD_UDP)
    {
        uv_handle_set_data((uv_handle_t *)m_targetTcp, this);
        SendCMDResp(m_cmdSendBuf, SOCKS5_VERSION, SOCKS5_SUCCEEDED, 0, SOCKS5_ATYP_IPV4, g_PublicIP, ((struct sockaddr_in *)&m_ss)->sin_port);
    }
}